package mqtt5;

import cn.com.haiyunhui.common.core.utils.ApplicationEngine;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * @Title: Mqtt5Client
 * @Package: cn.mis.util.mqtt5
 * @Description: mqtt5客户端
 * @Author: ChenLi
 * @Date: 2023/2/14 - 10:58
 */

@Data
@Slf4j
public class Mqtt5Client {
    private Map<String,String> topics = new HashMap<>();
    private MqttClient instance = null;
    private Emqx5Config mqttConfig;
    private PubCallback pushCallback = ApplicationEngine.getBean(PubCallback.class);

    public Mqtt5Client() {
    }
    public Mqtt5Client(Emqx5Config mqttConfig) {
        this.mqttConfig = mqttConfig;
    }

    /**
     * 返回MqttClient的单例
     *
     * @return
     */
    public MqttClient getInstance() {
        try {
            if (instance == null) {
                instance = new MqttClient(getRandomUrl(), mqttConfig.getClientId() + System.currentTimeMillis(), new MemoryPersistence());
                mqttConfig.init();
                if (StringUtils.isNotBlank(mqttConfig.getCallback())) {
                    Class clazz = Class.forName(mqttConfig.getCallback());
                    PubCallback callback = null;
                    try {
                        callback = (PubCallback) clazz.newInstance();
                    } catch (InstantiationException e) {
                        e.printStackTrace();
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                    if (callback != null) {
                        this.pushCallback = callback;
                    }
                }
                pushCallback.setClientName(mqttConfig.getClientName());
                instance.setCallback(pushCallback);
            }
        } catch (MqttException e) {
            e.printStackTrace();
            log.error(e.getMessage());
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            log.error(e.getMessage());
        }
        return instance;
    }

    public void init() {
        this.connect();
        this.subscribe();
    }

    public void reconnect() {
        try {
            instance.reconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
        if (isConnected()) {
            log.info("重连成功：MQTT客户端：" + mqttConfig.getClientName() + "启动成功！");
            this.subscribesed();
        }
    }

    public boolean connect() {
        try {
            this.getInstance().connect(mqttConfig.getOptions());
            log.debug("MQTT客户端：" + mqttConfig.getClientName() + "启动成功！");
            System.out.println("MQTT客户端：" + mqttConfig.getClientName() + "启动成功！");
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("MQTT客户端：" + mqttConfig.getClientName() + "启动失败！");
            log.error(e.getMessage());
        }
        return this.isConnected();
    }

    public void subscribe() {
        String[] topics = mqttConfig.getEventReportTopic().split(",");
        for (String topic : topics) {
            subscribe(topic, 1);
        }
    }

    public void subscribe(String topic, int qos) {
        topics.put(topic, topic);
        sub(topic, qos);
    }

    private void sub(String topic, int qos) {
        try {
            this.getInstance().subscribe(topic, qos);
            topics.put(topic, topic);
            log.debug("MQTT客户端：" + mqttConfig.getClientName() + "订阅成功！topic = " + topic);
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("MQTT客户端：" + mqttConfig.getClientName() + "订阅失败！topic = " + topic);
            log.error(e.getMessage());
        }
    }

    public void subscribesed(){
        for (String topic : topics.values()) {
            sub(topic, 1);
        }
    }

    public void unsubscribe(String topic){
        try {
            topics.remove(topic);
            this.getInstance().unsubscribe(topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public boolean disconnect(){
        try {
            instance.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
        return this.isConnected();
    }
    public boolean close() {
        try {
            disconnect();
            instance.close(false);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        return this.isConnected();
    }

    /**
     * 发布消息,默认消息质量为0
     *
     * @param content 消息的文本
     * @param topic   发送的主题
     */
    public boolean sendMessege(String topic, String content) {
        return sendMessege(topic, content, 0, false);
    }

    public boolean sendMessege(String topic, String content, int Qos, boolean retained) {
        try {
            this.getInstance().publish(topic, content.getBytes(), Qos, retained);
            log.debug("MQTT客户端：" + mqttConfig.getClientName() + "消息发布成功！topic = " + topic + " ### content = " + content);
            return true;
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("MQTT客户端：" + mqttConfig.getClientName() + "消息发布失败！topic = " + topic + " ### content = " + content);
            log.error(e.getMessage());
            return false;
        }
    }

    private String getRandomUrl(){
        try {
            String brokerUrl = mqttConfig.getBrokerUrl();
            String[] urls = brokerUrl.split(",");
            Random random = new Random();
            int index = random.nextInt(urls.length);
            return urls[index];
        }catch (Exception e){
            return mqttConfig.getBrokerUrl().split(",")[0];
        }
    }

    public boolean isConnected() {
        return this.getInstance().isConnected();
    }

}
